package com.yhou.demo.search.esbase;

import com.alibaba.fastjson.JSON;
import com.yhou.demo.search.common.Column;
import com.yhou.demo.search.enums.ESFieldTypeEnum;
import com.yhou.demo.search.enums.EsConstant;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.search.SearchHit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;


@Slf4j
@Getter
@Setter
public abstract class AbstractESHandler implements SearchBaseService {
    private static Logger log = LoggerFactory.getLogger(AbstractESHandler.class);

    @Autowired
    public TransportClient client;

    private String indexName;

    private String indexType;

    private Class modelClass;

    private Integer primaryShardNum;

    private AbstractESHandler(String indexName, String indexType) {
        this.indexName = indexName;
        this.indexType = indexType;
    }

    public AbstractESHandler(String indexName, String indexType, Class modelClass) {
        this(indexName, indexType);
        this.modelClass = modelClass;
    }

    public AbstractESHandler(String indexName, String indexType, Integer primaryShardNum, Class modelClass) {
        this(indexName, indexType);
        this.primaryShardNum = primaryShardNum;
        this.modelClass = modelClass;
    }
    /***
     * build model to ES doc
     *
     * 如果字段需要特殊处理，请在getter里做相应的业务处理
     */
    public <T> XContentBuilder generateIndexObject(T model) {
        try {
            XContentBuilder builder = jsonBuilder().startObject();
            Field[] fields = model.getClass().getDeclaredFields();
            for (Field field : fields) {
                //设置是否允许访问
                field.setAccessible(true);
                String classType = field.getType().getCanonicalName();
                try {
                    //获取getter方法并通过getter获取私有属性
                    String getterName = (new StringBuilder()).append(Character.toUpperCase(field.getName().charAt(0))).append(field.getName().substring(1)).toString();
                    getterName = "get" + getterName;
                    Method method;
                    try {
                        method = model.getClass().getMethod(getterName);
                    } catch (NoSuchMethodException e) {
                        log.warn("未找到方法{}.{}()", model.getClass().getName(), getterName);
                        continue;
                    }
                    Object value = method.invoke(model);

                    if (null == value) {
                        continue;
                    }

                    switch (classType) {
                        case EsConstant.LONG:
                        case EsConstant.INTEGER:
                        case EsConstant.STRING:
                        case EsConstant.FLOAT:
                        case EsConstant.DOUBLE:
                            builder.field(field.getName(), value);
                            break;
                        case EsConstant.DATE:
                            //日期处理注释
                            Column annotationColumn = field.getAnnotation(Column.class);
                            SimpleDateFormat simpleDateFormat;
                            if (null == annotationColumn) {
                                simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            } else {
                                String format = annotationColumn.ESType().format();
                                simpleDateFormat = new SimpleDateFormat(format);
                            }
                            builder.field(field.getName(), field.get(model) == null ? null : simpleDateFormat.format(field.get(model)));
                            break;
                        default:
                            break;
                    }
                } catch (Exception e) {
                    log.error("ES-Exception:: generateIndexObject classType={}", classType, e);
                }
            }
            builder.endObject();
            return builder;
        } catch (Exception e) {
            log.error("ES-Exception::generateIndexObject()!!!", e);
            throw new RuntimeException("ES-Exception::generateIndexObject() Exception!!!" + e.getMessage());
        }
    }

    /**
     * 判断指定的索引名是否存在
     *
     * @param indexName 索引名
     * @return 存在：true; 不存在：false;
     */
    private boolean isExistsIndex(String indexName) {
        IndicesExistsResponse response =
                client.admin().indices().exists(
                        new IndicesExistsRequest().indices(indexName)).actionGet();
        return response.isExists();
    }

    /**
     * 判断指定的索引的类型是否存在
     *
     * @param indexName 索引名
     * @param indexType 索引类型
     * @return 存在：true; 不存在：false;
     */
    private boolean isExistsType(String indexName, String indexType) {
        TypesExistsResponse response =
                client.admin().indices()
                        .typesExists(new TypesExistsRequest(new String[]{indexName}, indexType))
                        .actionGet();
        return response.isExists();
    }

    @PostConstruct
    public boolean checkIndexAndType() {
        boolean checkResult = false;
        try {
            if (!isExistsIndex(indexName)) {
                //创建index
                XContentBuilder settings = setDocSettings();
                log.info("XContentBuilder  settings= " + settings.string());
                CreateIndexResponse response = client.admin().indices().prepareCreate(indexName).setSettings(settings).execute().actionGet();

                if (!response.isShardsAcknowledged()) {
                    return false;
                }
            }
            if (!isExistsType(indexName, indexType)) {
                //创建mapping
                XContentBuilder builder = generateIndexMapping();
                if (null == builder) {
                    return false;
                }
                PutMappingRequest putMappingRequest = Requests.putMappingRequest(indexName).type(indexType).source(builder);
                PutMappingResponse mappingResponse = client.admin().indices().putMapping(putMappingRequest).actionGet();
                if (mappingResponse.isAcknowledged()) {
                    checkResult = true;
                }
            }

            log.info("index已经存在,无需再次创建! index={},type={}", indexName, indexType);
        } catch (Exception e) {
            log.error("创建index AND mapping exception! index={},type={},mapping={}", indexName, indexType, generateIndexMapping(), e);
        }
        return checkResult;
    }

    public XContentBuilder setDocSettings() {
        XContentBuilder settings = null;
        try {
            settings = XContentFactory.jsonBuilder()
                    .startObject();
            //分片数量
            if (this.primaryShardNum == null) {
                settings.field(EsConstant.FIELD_SHARDS, EsConstant.NUMBER_OF_SHARDS);
            } else {
                settings.field(EsConstant.FIELD_SHARDS, this.primaryShardNum);
            }
            //副本数量
            settings.field(EsConstant.FIELD_REPLICAS, EsConstant.NUMBER_OF_REPLICAS)
                    .field(EsConstant.FIELD_INTERVAL, EsConstant.REFRESH_INTERVAL)
                    .endObject();
        } catch (IOException e) {
            log.error("ES-Exception::setDocSettings !!!", e);
        }
        return settings;
    }

    /**
     * 创建ES Mapping
     *
     * {
     * 　  "type" : "text", #是数据类型一般文本使用text(可分词进行模糊查询)；keyword无法被分词(不需要执行分词器)，用于精确查找
     *
     *     "analyzer" : "ik_max_word", #指定分词器，一般使用最大分词：ik_max_word
     *
     *     "normalizer" : "normalizer_name", #字段标准化规则；如把所有字符转为小写；具体如下举例
     *
     *     "boost" : 1.5, #字段权重；用于查询时评分，关键字段的权重就会高一些，默认都是1；另外查询时可临时指定权重
     *
     *     "coerce" : true, #清理脏数据：1，字符串会被强制转换为整数 2，浮点数被强制转换为整数；默认为true
     *
     *     "copy_to" : "field_name", #自定_all字段；指定某几个字段拼接成自定义；具体如下举例
     *
     *     "doc_values" : true, #加快排序、聚合操作，但需要额外存储空间；默认true，对于确定不需要排序和聚合的字段可false
     *
     *     "dynamic" : true, #新字段动态添加 true:无限制 false:数据可写入但该字段不保留 'strict':无法写入抛异常
     *
     *     "enabled" : true, #是否会被索引，但都会存储;可以针对一整个_doc
     *
     *     "fielddata" : false, #针对text字段加快排序和聚合（doc_values对text无效）；此项官网建议不开启，非常消耗内存
     *
     *     "eager_global_ordinals": true, #是否开启全局预加载,加快查询；此参数只支持text和keyword，keyword默认可用，而text需要设置fielddata属性
     *
     *     "format" : "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" ,#格式化 此参数代表可接受的时间格式 3种都接受
     *
     *     "ignore_above" : 100, #指定字段索引和存储的长度最大值，超过最大值的会被忽略
     *
     *     "ignore_malformed" : false ,#插入文档时是否忽略类型 默认是false 类型不一致无法插入
     *
     *     "index_options" : "docs" ,
     *     # 4个可选参数
     *     # docs（索引文档号）,
     *     # freqs（文档号 + 词频），
     *     # positions（文档号 + 词频 + 位置，通常用来距离查询），
     *     # offsets（文档号 + 词频 + 位置 + 偏移量，通常被使用在高亮字段）
     *     # 分词字段默认是position，其他的默认是docs
     *
     *     "index" : true, #该字段是否会被索引和可查询 默认true
     *
     *     "fields": {"raw": {"type": "keyword"}} ,#可以对一个字段提供多种索引模式，使用text类型做全文检索，也可使用keyword类型做聚合和排序
     *
     *     "norms" : true, #用于标准化文档，以便查询时计算文档的相关性。建议不开启
     *
     *     "null_value" : "NULL", #可以让值为null的字段显式的可索引、可搜索
     *
     *     "position_increment_gap" : 0 ,#词组查询时可以跨词查询 既可变为分词查询 默认100
     *
     *     "properties" : {}, #嵌套属性，例如该字段是音乐，音乐还有歌词，类型，歌手等属性
     *
     *     "search_analyzer" : "ik_max_word" ,#查询分词器;一般情况和analyzer对应
     *
     *     "similarity" : "BM25",#用于指定文档评分模型，参数有三个：
     *     # BM25 ：ES和Lucene默认的评分模型
     *     # classic ：TF/IDF评分
     *     # boolean：布尔模型评分
     *
     *     "store" : true, #默认情况false,其实并不是真没有存储，_source字段里会保存一份原始文档。
     *     # 在某些情况下，store参数有意义，比如一个文档里面有title、date和超大的content字段，如果只想获取title和date
     *
     *     "term_vector" : "no" #默认不存储向量信息，
     *     # 支持参数yes（term存储），
     *     # with_positions（term + 位置）,
     *     # with_offsets（term + 偏移量），
     *     # with_positions_offsets(term + 位置 + 偏移量)
     *     # 对快速高亮fast vector highlighter能提升性能，但开启又会加大索引体积，不适合大数据量用
     * }
     *
     * normalizer举例：
     * {
     *   "settings": {
     *     "analysis": {
     *       "normalizer": {
     *         "my_normalizer": {
     *           "type": "custom",
     *           "char_filter": [],
     *           "filter": ["lowercase", "asciifolding"]
     *         }
     *       }
     *     }
     *   },
     *   "mappings": {
     *     "type": {
     *       "properties": {
     *         "foo": {
     *           "type": "keyword",
     *           "normalizer": "my_normalizer"
     *         }
     *       }
     *     }
     *   }
     * }
     *
     * copy_to举例：
     * {
     *   "mappings": {
     *     "my_type": {
     *       "properties": {
     *         "first_name": {
     *           "type": "text",
     *           "copy_to": "full_name"
     *         },
     *         "last_name": {
     *           "type": "text",
     *           "copy_to": "full_name"
     *         },
     *         "full_name": {
     *           "type": "text"
     *         }
     *       }
     *     }
     *   }
     * }
     *
     * @return
     * @throws IOException
     */
    public XContentBuilder generateIndexMapping() {
        XContentBuilder builder = null;
        try {
            builder = jsonBuilder().startObject();
            builder.startObject(String.valueOf(getIndexType()));
            builder.startObject(EsConstant.FIELD_PROPERTIES);
            Field[] fields = modelClass.getDeclaredFields();
            for (Field field : fields) {
                //设置是否允许访问，不是修改原来的访问权限修饰词。
                field.setAccessible(true);
                //处理注释字段
                if (field.isAnnotationPresent(Column.class)) {
                    handleAnnotationFieldMapping(builder, field);
                    continue;
                }
                //处理默认字段
                handleDefaultFieldMapping(builder, field);
            }
            builder.endObject();
            builder.endObject();
            builder.endObject();
        } catch (Exception e) {
            log.error("ES-Exception::generateIndexMapping()!!!", e);
        }
        return builder;
    }

    /**
     * 处理没有Column注解或者Column.ESType为DEFAULT的字段
     *
     * @param builder
     * @param field
     * @throws IOException
     */
    private void handleDefaultFieldMapping(XContentBuilder builder, Field field) throws IOException {
        String classType = field.getType().getCanonicalName();
        switch (classType) {
            case EsConstant.LONG:
                builder.startObject(field.getName()).field(EsConstant.TYPE, EsConstant.ES_LONG).endObject();
                break;
            case EsConstant.STRING:
                builder.startObject(field.getName()).field(EsConstant.TYPE, EsConstant.ES_KEYWORD).endObject();
                break;
            case EsConstant.INTEGER:
                builder.startObject(field.getName()).field(EsConstant.TYPE, EsConstant.ES_INTEGER).endObject();
                break;
            case EsConstant.FLOAT:
                builder.startObject(field.getName()).field(EsConstant.TYPE, EsConstant.ES_FLOAT).endObject();
                break;
            case EsConstant.DOUBLE:
                builder.startObject(field.getName()).field(EsConstant.TYPE, EsConstant.ES_DOUBLE).endObject();
                break;
            case EsConstant.DATE:
                builder.startObject(field.getName()).field(EsConstant.TYPE, EsConstant.ES_DATE).
                        field(EsConstant.FIELD_FORMAT, ESFieldTypeEnum.DATETIME.format()).endObject();
                break;
            default:
                break;
        }
    }

    /**
     * 处理Column注解的字段
     *
     * @param builder
     * @param field
     * @throws IOException
     */
    private void handleAnnotationFieldMapping(XContentBuilder builder, Field field) throws IOException {
        Column annotationColumn = field.getAnnotation(Column.class);
        switch (annotationColumn.ESType()) {
            case DATE:
            case DATETIME:
            case GEO_POINT:
                builder.startObject(field.getName());
                builder.field(EsConstant.TYPE, annotationColumn.ESType().type());
                if (StringUtils.isNotEmpty(annotationColumn.ESType().format())) {
                    builder.field(EsConstant.FIELD_FORMAT, annotationColumn.ESType().format());
                }
                builder.endObject();
                break;
            default:
                handleDefaultFieldMapping(builder, field);
                break;
        }
    }

    @Override
    public boolean delete(Serializable id) {
        if (id == null) {
            log.info("删除{}索引参数id为空！");
            return true;
        }
        DeleteRequest deleteRequest = new DeleteRequest();
        deleteRequest.id(String.valueOf(id));
        deleteRequest.index(getIndexName());
        deleteRequest.type(getIndexType());
        try {
            DeleteResponse deleteResponse = client.delete(deleteRequest).actionGet();
            log.info("删除{}索引请求response={}", getIndexName(), deleteResponse);
            if (deleteResponse.getResult() == DeleteResponse.Result.DELETED) {
                log.info("删除{}索引成功！id={}", getIndexName(), id);
                return true;
            } else if (deleteResponse.getResult() == DeleteResponse.Result.NOT_FOUND) {
                log.warn("删除{}索引不存在！id={}", getIndexName(), id);
                return true;
            }
        } catch (Exception e) {
            log.error("删除{}索引异常,id={}", getIndexName(), id, e);
        }
        return false;
    }
    public   <T> List<T> parseHits2Class(SearchHit[] hitArr,Class<T> calss) {
        return Arrays.stream(hitArr).filter(Objects::nonNull).map(hit -> {
            Map<String, Object> map = hit.getSourceAsMap();
            String json = JSON.toJSONString(map);
            return JSON.parseObject(json, calss);
        }).collect(Collectors.toList());
    }

    public String getIndexName() {
        return indexName;
    }

    public void setIndexName(String indexName) {
        this.indexName = indexName;
    }

    public String getIndexType() {
        return indexType;
    }

    public void setIndexType(String indexType) {
        this.indexType = indexType;
    }

    public Class getModelClass() {
        return modelClass;
    }

    public void setModelClass(Class modelClass) {
        this.modelClass = modelClass;
    }

    public Integer getPrimaryShardNum() {
        return primaryShardNum;
    }

    public void setPrimaryShardNum(Integer primaryShardNum) {
        this.primaryShardNum = primaryShardNum;
    }
}
